package com.wk.masterNode;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import com.wk.job.Job;
import com.wk.job.Task;

/***
 * Master 包括下列几个部分：
 *  1. HeartThread 接收心跳包。单独的线程。如果心跳包中包含任务信息则更改主节点任务队列中的Job的执行情况。
 *  2. Creator 读取数据库中的记录，创建Job并放入同步队列LinkedBlockingQueue中。
 *  	2.1 tasksQueue 记录一条数据库记录，一个task对应多个job
 *  3. Consumer 在jobsQueue中获取一个Job，按照省份与机器的对应表发给相应的处理机器。把分发的所有任务存储在一个distributedJobs中。
 *  4. jobsQueue 记录所有待分配的任务。
 *  5. distributedJobs 记录所有的已经分配给计算节点的job。记录当前的任务完成百分比。如果任务已经完成或者超时，则把任务写入数据库中并从ConcurrentHashMap中删除该任务。
 *  
 * @author wkhua
 *
 */

public class Master {
	
	//记录所有在数据库中读取到的Task
	private ConcurrentHashMap<Integer,Task> tasks = new ConcurrentHashMap<>();
	//记录由Task分解出的job
	private LinkedBlockingQueue<Job> jobsQueue = new LinkedBlockingQueue<>(10);
	//记录所有成功被分发到计算节点的job
	private ConcurrentHashMap<String,Job> distributedJobs = new ConcurrentHashMap<>();
	
	public Master(){
		
	}
	public boolean initMaster() {
		JobStateHeartbeatHandlerBuilder heartbeatHandlerBuilder = new JobStateHeartbeatHandlerBuilder();
		try {
			heartbeatHandlerBuilder.listen(8086);
		} catch (IOException e) {
			e.printStackTrace();
			return false;
		}
		heartbeatHandlerBuilder.loadSlaves();
		heartbeatHandlerBuilder.dumpObservedSlaves();
		
		Thread creator = new Thread(new Creator(jobsQueue,tasks),"生产者线程");
		Thread consumer = new Thread(new Consumer(jobsQueue,distributedJobs),"消费者线程");
		
		heartbeatHandlerBuilder.makeHandler(distributedJobs, tasks, 5, 3).start();
		creator.start();
		consumer.start();
		
		return true;
	}
	public static void main(String[] args){
		Master master = new Master();
		boolean foo = master.initMaster();
		System.out.println(foo);
	}
}
